package cn.lzd.mr.mroverride.distributedCache;

import org.apache.commons.collections.map.HashedMap;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Iterator;
import java.util.Map;

/**
 * nohup hadoop jar hive.jar /inzdl/company2/company2 /outzdl/all/demo3 hdfs://172.16.64.50:9000/inzdl/table/company/industry.txt cn.lzd.all.company.demo3.Demo3 &
 * Created by lzd on 2018/12/12.
 * 过滤不合法的数据，合法ok,不合法no
 * 依据：如果公司有id，公司所属省份在34个省份内，或是'|N'或""中，合法
 * 没有id或者省份为中文，不合法
 */
public class UpDemo3 {
    public static class MyMapper extends Mapper<Object, Text, Text, Text> {

        Text outKey = new Text();
        Text outVal = new Text();
        Map<String, String> map = new HashedMap();

        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            //线上
            BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream("industry.txt"), "UTF-8"));

            String line;
            while (StringUtils.isNotBlank(line = reader.readLine())) {
                String[] codes = line.split("\\|");
                map.put(codes[0], codes[codes.length - 1]);
            }
        }

        @Override
        protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            String val = value.toString();
            String[] arr = val.split("\\|");

            try {
                Integer id = Integer.valueOf(arr[0]);
                String base = arr[1];
                if (StrUtil.checkBase(base)) {
                    outKey.set("ok" + id);

                    //加大行业编码
                    String code = arr[arr.length - 1];
                    if (!StrUtil.isBlank(code)) {
                        String cate = map.get(code);
                        val = val + "|" + cate;
                    }

                } else {
                    outKey.set("no" + System.nanoTime());
                }
            } catch (NumberFormatException e) {
                outKey.set("no" + System.nanoTime());
            }
            outVal.set(val);
            context.write(outKey, outVal);
        }
    }

    public static class MyReduce extends Reducer<Text, Text, NullWritable, Text> {
        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            Iterator<Text> iterator = values.iterator();
            while (iterator.hasNext()) {
                Text next = iterator.next();
                context.write(NullWritable.get(), next);
            }
        }
    }

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException, URISyntaxException {

        Configuration configuration = new Configuration();

        Job job = Job.getInstance(configuration, "all-company-demo3");

        job.setJarByClass(UpDemo3.class);

        job.setMapperClass(UpDemo3.MyMapper.class);
        job.setReducerClass(UpDemo3.MyReduce.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(Text.class);

        job.setPartitionerClass(Demo3Partition.class);
        job.setNumReduceTasks(201);

        //"/G:\\mr_profile/alltest"
        //"/G:\\mr_profile/out"
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        //线上写法
        job.addCacheFile(new URI("hdfs://172.16.**.50:9000/inzdl/table/company/industry.txt"));

        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}
